/**
 * @file aiot_ra_api.c
 * @brief remote-access模块的API接口实现, 提供远程隧道的能力
 *
 * @copyright Copyright (C) 2015-2020 Alibaba Group Holding Limited
 *
 */
#include <stdio.h>
#include "core_stdinc.h"
#include "core_log.h"
#include "aiot_sysdep_api.h"
#include "aiot_state_api.h"
#include "aiot_ra_api.h"
#include "aiot_ra_private.h"
#include "aiot_mqtt_api.h"
#include "core_mqtt.h"
#include "z_logger.h"

static const char *FMT_TOPIC_SWITCH = "/sys/%s/%s/secure_tunnel/notify";

void *aiot_ra_init(void) {
    aiot_sysdep_portfile_t *sysdep = NULL;
    ra_handle_t *ra_handle = NULL;
    sysdep = aiot_sysdep_get_portfile();
    if (NULL == sysdep) {
        return NULL;
    }

    ra_handle = sysdep->core_sysdep_malloc(sizeof(ra_handle_t), REMOTE_ACCESS_MODULE_NAME);
    if (ra_handle == NULL) {
        return NULL;
    }
    memset(ra_handle, 0, sizeof(ra_handle_t));
    ra_handle->sysdep = sysdep;
    CORE_INIT_LIST_HEAD(&ra_handle->local_services.service_list);
    return ra_handle;
}

static void _release_all_service_info(LOCAL_SERVICES_S *local_services) {
    aiot_sysdep_portfile_t *sysdep = NULL;
    sysdep = aiot_sysdep_get_portfile();
    if (NULL == sysdep) {
        return;
    }
    LOCAL_SERVICE_NODE_S *item = NULL, *next = NULL;
    core_list_for_each_entry_safe(item, next, &local_services->service_list, node, LOCAL_SERVICE_NODE_S) {
        core_list_del(&item->node);
        sysdep->core_sysdep_free(item);
    }
}

int32_t aiot_ra_setopt(void *handle, aiot_ra_option_t option, void *data) {
    ra_handle_t *ra_handle = (ra_handle_t *) handle;

    if (ra_handle == NULL || data == NULL) {
        return STATE_USER_INPUT_NULL_POINTER;
    }
    if (option >= AIOT_RAOPT_MAX) {
        return STATE_USER_INPUT_OUT_RANGE;
    }

    switch (option) {
        case AIOT_RAOPT_MQTT_HANDLE:
            ra_handle->mqtt_handle = data;
            break;
        case AIOT_RAOPT_EVENT_HANDLER:
            ra_handle->event_handle = (aiot_ra_event_handler_t) data;
            break;
        case AIOT_RAOPT_USERDATA:
            ra_handle->userdata = data;
            break;
        case AIOT_RAOPT_NETWORK_CRED: {
            aiot_sysdep_portfile_t *sysdep = aiot_sysdep_get_portfile();
            if (ra_handle->cred != NULL) {
                sysdep->core_sysdep_free(ra_handle->cred);
                ra_handle->cred = NULL;
            }
            ra_handle->cred = sysdep->core_sysdep_malloc(sizeof(aiot_sysdep_network_cred_t), REMOTE_ACCESS_MODULE_NAME);
            if (ra_handle->cred != NULL) {
                memset(ra_handle->cred, 0, sizeof(aiot_sysdep_network_cred_t));
                memcpy(ra_handle->cred, data, sizeof(aiot_sysdep_network_cred_t));
            } else {
                return STATE_SYS_DEPEND_MALLOC_FAILED;
            }
        }
            break;
        case AIOT_RAOPT_ADD_SERVICE: {
            aiot_ra_service_t *service = (aiot_ra_service_t *) data;
            LOCAL_SERVICE_NODE_S *service_node = ra_handle->sysdep->core_sysdep_malloc(sizeof(LOCAL_SERVICE_NODE_S),
                                                                                       REMOTE_ACCESS_MODULE_NAME);
            memset(service_node, 0, sizeof(LOCAL_SERVICE_NODE_S));
            memcpy(service_node->type, service->type, strlen(service->type));
            memcpy(service_node->ip, service->ip, strlen(service->ip));
            service_node->port = service->port;

            core_list_add(&service_node->node, &ra_handle->local_services.service_list);
        }
            break;
        default:
            break;
    }

    return STATE_SUCCESS;
}

int32_t aiot_ra_deinit(void **handle) {
    int i = 0;
    if (NULL == handle || NULL == *handle) {
        return STATE_USER_INPUT_OUT_RANGE;
    }

    ra_handle_t *ra_handle = *(ra_handle_t **) handle;
    aiot_sysdep_portfile_t *sysdep = aiot_sysdep_get_portfile();
    if (NULL == sysdep) {
        return STATE_REMOTE_ACCESS_FAILED;
    }

    if (ra_handle->cred != NULL) {
        sysdep->core_sysdep_free(ra_handle->cred);
        ra_handle->cred = NULL;
    }

    _release_all_service_info(&ra_handle->local_services);
    for (i = 0; i < MAX_PROXY_NUM; i++) {
        if (ra_handle->proxy_info[i] != NULL) {
            deinit_remote_proxy_resource(&ra_handle->proxy_info[i]);
        }
    }
    sysdep->core_sysdep_free(*handle);
    return STATE_SUCCESS;
}

void
remote_proxy_event_handle(ra_handle_t *ra_handle, aiot_ra_event_type type, REMOTE_PROXY_INFO_S *remote_proxy_info) {
    aiot_ra_event_t event;
    event.type = type;
    strncpy(event.tunnel_id, remote_proxy_info->tunnel_id, sizeof(remote_proxy_info->tunnel_id));
    ra_handle->event_handle(ra_handle, &event, ra_handle->userdata);
}

static void _switch_topic_handler(void *handle, const aiot_mqtt_recv_t *packet, void *userdata) {
    char *data = NULL;
    uint32_t data_strlen = 0;
    int32_t res, i = 0;
    ra_handle_t *ra_handle = (ra_handle_t *) userdata;
    REMOTE_PROXY_INFO_S *remote_proxy_info = NULL;
    char tunnel_id[128];
    if (packet->type == AIOT_MQTTRECV_PUB) {
        int32_t payload_len = packet->data.pub.payload_len;
        core_log2(aiot_sysdep_get_portfile(), STATE_REMOTE_ACCESS_BASE, "_switch_topic_handler payload:%.*s\r\n",
                  &payload_len, packet->data.pub.payload);

        memset(tunnel_id, 0, sizeof(tunnel_id));
        res = core_json_value((const char *) packet->data.pub.payload, packet->data.pub.payload_len, "tunnel_id",
                              strlen("tunnel_id"), &data, &data_strlen);
        if (res == STATE_SUCCESS) {
            memcpy(tunnel_id, data, data_strlen);
        } else {
            return;
        }

        /* 查看隧道是否已经打开了 */
        for (i = 0; i < MAX_PROXY_NUM; i++) {
            if (ra_handle->proxy_info[i] != NULL &&
                0 == strncmp(ra_handle->proxy_info[i]->tunnel_id, tunnel_id, strlen(tunnel_id))) {
                remote_proxy_info = ra_handle->proxy_info[i];
                break;
            }
        }

        /* 解析隧道的操作 */
        res = core_json_value((const char *) packet->data.pub.payload, packet->data.pub.payload_len, "operation",
                              strlen("operation"), &data, &data_strlen);
        if (res < STATE_SUCCESS) {
            return;
        }
        if (0 == strncmp(data, "connect", data_strlen)) {
            if (remote_proxy_info == NULL) {
                //初始化资源
                remote_proxy_info = ra_handle->sysdep->core_sysdep_malloc(sizeof(REMOTE_PROXY_INFO_S), "RA");
                if (0 != init_remote_proxy_resource(remote_proxy_info)) {
                    core_log(aiot_sysdep_get_portfile(), STATE_REMOTE_ACCESS_BASE,
                             "init remote proxy resource error!\r\n");
                }
                memcpy(remote_proxy_info->tunnel_id, tunnel_id, strlen(tunnel_id));
                remote_proxy_info->cloud_channel_params.cred = ra_handle->cred;
                remote_proxy_info->cloud_channel_params.local_services = &ra_handle->local_services;
            } else {
                core_log1(aiot_sysdep_get_portfile(), STATE_REMOTE_ACCESS_BASE, "reopen tunnel_id %s !\r\n", tunnel_id);
                return;
            }

        } else if (0 == strncmp(data, "close", data_strlen)) {
            if (remote_proxy_info != NULL) {
                remote_proxy_event_handle(ra_handle, AIOT_RA_EVT_CLOSE_WEBSOCKET, ra_handle->proxy_info[i]);
                ra_handle->proxy_info[i]->remote_proxy_channel_switch = 0;
                ra_handle->proxy_info[i]->has_switch_event = 1;
                ra_handle->proxy_info[i]->status = 0;
                ra_handle->channel_num--;
            } else {
                core_log1(aiot_sysdep_get_portfile(), STATE_REMOTE_ACCESS_BASE, "close tunnel_id %s not found!\r\n",
                          tunnel_id);
            }

            return;
        }

        res = core_json_value((const char *) packet->data.pub.payload, packet->data.pub.payload_len, "host",
                              strlen("host"), &data, &data_strlen);
        if (res == STATE_SUCCESS) {
            memset(remote_proxy_info->cloud_channel_params.cloud_host, 0,
                   sizeof(remote_proxy_info->cloud_channel_params.cloud_host));
            memcpy(remote_proxy_info->cloud_channel_params.cloud_host, data, data_strlen);
        }

        res = core_json_value((const char *) packet->data.pub.payload, packet->data.pub.payload_len, "port",
                              strlen("port"), &data, &data_strlen);
        if (res == STATE_SUCCESS) {
            memcpy(remote_proxy_info->cloud_channel_params.cloud_port, data, data_strlen);
        }

        res = core_json_value((const char *) packet->data.pub.payload, packet->data.pub.payload_len, "path",
                              strlen("path"), &data, &data_strlen);
        if (res == STATE_SUCCESS) {
            memcpy(remote_proxy_info->cloud_channel_params.cloud_path, data, data_strlen);
        }

        res = core_json_value((const char *) packet->data.pub.payload, packet->data.pub.payload_len, "token",
                              strlen("token"), &data, &data_strlen);
        if (res == STATE_SUCCESS) {
            memcpy(remote_proxy_info->cloud_channel_params.token, data, data_strlen);
        }

        remote_proxy_info->remote_proxy_channel_switch = 1;
        remote_proxy_info->has_switch_event = 1;
        remote_proxy_info->status = 1;
        for (i = 0; i < MAX_PROXY_NUM; i++) {
            if (ra_handle->proxy_info[i] == NULL) {
                ra_handle->proxy_info[i] = remote_proxy_info;
                ra_handle->channel_num++;
                remote_proxy_event_handle(ra_handle, AIOT_RA_EVT_OPEN_WEBSOCKET, remote_proxy_info);
                break;
            }
        }
    }
}

static int32_t _sub_switch_topic(void *handle) {
    int32_t res = STATE_SUCCESS;
    ra_handle_t *ra_handle = (ra_handle_t *) handle;
    char cloud_switch_commond_topic[CLOUD_TOPIC_LEN_MAX];
    char *pk = core_mqtt_get_product_key(ra_handle->mqtt_handle);
    char *dn = core_mqtt_get_device_name(ra_handle->mqtt_handle);
    snprintf(cloud_switch_commond_topic, CLOUD_TOPIC_LEN_MAX, FMT_TOPIC_SWITCH, pk, dn);
    res = aiot_mqtt_sub(ra_handle->mqtt_handle, cloud_switch_commond_topic, _switch_topic_handler, 1, handle);
    if (res < 0) {
        core_log(aiot_sysdep_get_portfile(), STATE_REMOTE_ACCESS_BASE, "aiot_mqtt_sub failed\r\n");
        return -1;
    }

    return res;
}

static int32_t _unsub_switch_topic(void *handle) {
    int32_t res = STATE_SUCCESS;
    ra_handle_t *ra_handle = (ra_handle_t *) handle;
    char cloud_switch_commond_topic[CLOUD_TOPIC_LEN_MAX];
    char *pk = core_mqtt_get_product_key(ra_handle->mqtt_handle);
    char *dn = core_mqtt_get_device_name(ra_handle->mqtt_handle);

    snprintf(cloud_switch_commond_topic, CLOUD_TOPIC_LEN_MAX, FMT_TOPIC_SWITCH, pk, dn);
    res = aiot_mqtt_unsub(ra_handle->mqtt_handle, cloud_switch_commond_topic);
    if (res < 0) {
        core_log(aiot_sysdep_get_portfile(), STATE_REMOTE_ACCESS_BASE, "aiot_mqtt_unsub failed\r\n");
        return -1;
    }

    return res;
}

void *remote_proxy_thread(void *params) {
    int i = 0;
    ra_handle_t *ra_handle = (ra_handle_t *) params;
    aiot_sysdep_portfile_t *sysdep = aiot_sysdep_get_portfile();
    if (sysdep == NULL) {
        return NULL;
    }

    _sub_switch_topic(params);
    core_log(aiot_sysdep_get_portfile(), STATE_REMOTE_ACCESS_BASE, "remote proxy thread start!\r\n ");
    //作为代理处理南北向数据，以下为状态机处理内容
    while (running) {
        for (i = 0; i < MAX_PROXY_NUM; i++) {
            REMOTE_PROXY_INFO_S *remote_proxy_info = ra_handle->proxy_info[i];
            if (remote_proxy_info != NULL) {
                REMOTE_PROXY_STATE_E last_state = remote_proxy_info->cloud_channel_state;
                remote_proxy_process(remote_proxy_info);
                if (last_state != remote_proxy_info->cloud_channel_state) {
                    if (last_state == CLOUD_CHANNEL_CLOSED) {
                        remote_proxy_event_handle(ra_handle, AIOT_RA_EVT_CONNECT, remote_proxy_info);
                    } else if (last_state == CLOUD_CHANNEL_CONNECTED) {
                        remote_proxy_event_handle(ra_handle, AIOT_RA_EVT_DISCONNECT, remote_proxy_info);
                        /* 如果用户主动关闭的话，删除远程代理 */
                        if (remote_proxy_info->status == 0) {
                            deinit_remote_proxy_resource(&ra_handle->proxy_info[i]);
                        }
                    }
                }
            }
        }

        if (ra_handle->channel_num == 0) {
            sysdep->core_sysdep_sleep(500);
        }

        //更新通道参数
        if (1 == ra_handle->pthread_exit_flag) {
            for (i = 0; i < MAX_PROXY_NUM; i++) {
                if (ra_handle->proxy_info[i] != NULL) {
                    deinit_remote_proxy_resource(&ra_handle->proxy_info[i]);
                }
            }

            break;
        }
    }

    _unsub_switch_topic(ra_handle);
    core_log(aiot_sysdep_get_portfile(), STATE_REMOTE_ACCESS_BASE, "remote proxy thread exit!\r\n");

    return NULL;
}

void *aiot_ra_start(void *handle) {
    ra_handle_t *ra_handle = (ra_handle_t *) handle;

    if (NULL == ra_handle) {
        return NULL;
    }

    ra_handle->result = STATE_SUCCESS;
    if (NULL == ra_handle->mqtt_handle) {
        ra_handle->result = STATE_REMOTE_ACCESS_MISSING_MQTT_HADNL;
    } else {
        remote_proxy_thread(handle);
        ra_handle->result = STATE_SUCCESS;
    }

    return &ra_handle->result;
}

int32_t aiot_ra_stop(void *handle) {
    ra_handle_t *ra_handle = (ra_handle_t *) handle;
    if (NULL == ra_handle) {
        return STATE_USER_INPUT_NULL_POINTER;
    }

    ra_handle->pthread_exit_flag = 1;
    return STATE_SUCCESS;
}

